/*
 * MIT License
 *
 * Copyright (c) 2021 cao.yong<cymsn81@126.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
package org.cy.websocket.stomp;

import org.cy.websocket.stomp.events.OnConnectedEvent;
import org.cy.websocket.stomp.events.OnDisconnectedEvent;
import org.cy.websocket.stomp.events.OnErrorEvent;
import org.cy.websocket.stomp.events.OnSubscribeEvent;
import org.cy.websocket.stomp.timer.TimedTaskScheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * StompClient
 * <p>
 * Created in 2021-05-08 21:44
 *
 * @author cao.yong
 */
public class StompClient {

    private static final Logger LOG = LoggerFactory.getLogger(StompClient.class);
    private static final String TASK_RECONNECT = "RECONNECT";

    private final StompHandler stompHandler;
    private final TimedTaskScheduler scheduler;
    /**
     * Keep Activation
     */
    private Constants.ActivationState state;

    /**
     * Global Headers;
     */
    private StompHeaders connectHeaders;
    private StompHeaders disconnectHeaders;

    // set 0 disable try reconnect
    private long reconnectDelay;

    public StompClient() {
        this.stompHandler = new StompHandler();
        this.scheduler = new TimedTaskScheduler();
        this.state = Constants.ActivationState.INACTIVE;
        this.connectHeaders = new StompHeaders();
        this.disconnectHeaders = new StompHeaders();
    }

    /**
     * 连接到服务器
     * @param headers {@link StompHeaders} STOMP 协议头
     */
    public void connect(StompHeaders headers) {
        if (Constants.ActivationState.INACTIVE.equals(this.state)) {
            connectHeaders.mergeAll(headers);
            this.activate();
        }
    }

    /**
     * 与服务器连接主动断开
     * @param headers STOMP 协议头信息
     */
    public void disconnect(StompHeaders headers) {
        if (Constants.ActivationState.ACTIVE.equals(this.state)) {
            this.changeState(Constants.ActivationState.DEACTIVATING);
            disconnectHeaders.mergeAll(headers);
            this.deactivate();
        }
    }

    /**
     * 是否已连接服务器
     * @return true or false
     */
    public boolean isConnected() {
        return this.stompHandler.isConnected();
    }

    /**
     * 发送消息到指定的目标通道
     *
     * @param destination String 目标通道
     * @param headers {@link StompHeaders} STOMP 协议头
     * @param content String 发送的内容
     */
    public void send(String destination, StompHeaders headers, String content) {
        this.stompHandler.send(destination, headers, content);
    }

    /**
     * 订阅目标通道
     * @param destination String 目标通道
     * @param headers {@link StompHeaders} STOMP 协议头
     * @param callback {@link OnSubscribeEvent} 订阅后接收通道消息
     * @return 订阅持久对象，可以进行取消订阅
     */
    public StompSubscription subscribe(String destination, StompHeaders headers, OnSubscribeEvent callback) {
        return this.stompHandler.subscribe(destination, headers, callback);
    }

    /**
     * 开始事务
     * <p>
     *  String txId = ...;
     *  StompTransaction transaction = client.begin(txId);
     *  ...
     *  transaction.commit();
     *  ...
     *  transaction.abort();
     * </p>
     * @param transactionId 事务标识
     * @return 事务对象
     */
    public StompTransaction begin(String transactionId) {
        return this.stompHandler.begin(transactionId);
    }

    /**
     * 应答
     *
     * @param messageId 消息标识
     * @param subscriptionId 订阅标识
     * @param headers {@link StompHeaders} STOMP 协议头
     */
    public void ack(String messageId, String subscriptionId, StompHeaders headers) {
        this.stompHandler.ack(messageId, subscriptionId, headers);
    }

    /**
     * 不应答
     *
     * @param messageId 消息标识
     * @param subscriptionId 订阅标识
     * @param headers {@link StompHeaders} STOMP 协议头
     */
    public void nack(String messageId, String subscriptionId, StompHeaders headers) {
        this.stompHandler.nack(messageId, subscriptionId, headers);
    }

    /**
     * 设置连接地址
     * @param url String
     */
    protected void  setBrokerUrl(String url) {
        this.stompHandler.setBrokerUrl(url);
    }

    /**
     * 设置连接时的 STOMP 协议头
     * @param connectHeaders {@link StompHeaders} STOMP 协议头
     */
    protected void setConnectHeaders(StompHeaders connectHeaders) {
        this.connectHeaders = connectHeaders;
    }

    /**
     * 设置断开时的 STOMP 协议头
     * @param disconnectHeaders {@link StompHeaders} STOMP 协议头
     */
    protected void setDisconnectHeaders(StompHeaders disconnectHeaders) {
        this.disconnectHeaders = disconnectHeaders;
    }

    /**
     * 设置心跳
     * @param incoming 由服务器主动发送心跳，0 关闭
     * @param outgoing 由客户端主动发送心跳，0 关闭
     */
    protected void setHeartbeat(long incoming, long outgoing) {
        this.stompHandler.setHeartbeat(new Heartbeat(incoming, outgoing));
    }

    /**
     * 异常断开后尝试重新连接，设置重连间隔时间
     * @param reconnectDelay long 间隔
     */
    protected void setReconnectDelay(long reconnectDelay) {
        this.reconnectDelay = reconnectDelay;
    }

    /**
     * 设置连接后的回调函数
     * @param onConnectedEvent 连接成功回调处理事件
     */
    protected void setOnConnected(OnConnectedEvent onConnectedEvent) {
        this.stompHandler.setOnConnectedEvent(response -> {
            onConnectedEvent.on(response);
            if (Constants.ActivationState.ACTIVE == this.state) {
                scheduler.stop(TASK_RECONNECT);
            }
        });
    }

    /**
     * 设置断开后的回调函数
     * @param onDisconnectedEvent 断开成功回调处理事件
     */
    protected void setOnDisconnected(OnDisconnectedEvent onDisconnectedEvent) {
        this.stompHandler.setOnDisconnectedEvent(onDisconnectedEvent);
        this.stompHandler.setOnClosedEvent((code, reason) -> {
            if (Constants.ActivationState.DEACTIVATING == this.state) {
                changeState(Constants.ActivationState.INACTIVE);
            }
            // 当客户端处于活动状态
            if (Constants.ActivationState.ACTIVE == this.state) {
                // 重试间隔大于0时，启动重连任务
                if (reconnectDelay > 0) {
                    scheduler.start(TASK_RECONNECT, () -> {
                        LOG.debug("STOMP Disconnected, Try Reconnect.");
                        this.activate();
                    }, reconnectDelay);
                }
            }
        });
    }

    /**
     * 连接中，发送错误的回调函数
     * @param onErrorEvent 发生错误后的回调处理事件
     */
    protected void setOnError(OnErrorEvent onErrorEvent) {
        this.stompHandler.setOnErrorEvent(ex -> {
            changeState(Constants.ActivationState.INACTIVE);
            onErrorEvent.on(ex);
        });
    }

    /**
     * 启动连接，并设置客户端活动状态
     */
    private void activate() {
        if (!isConnected()) {
            this.stompHandler.setSocket(StompWebSocketFactory.createStompWebSocket());
            this.stompHandler.start(this.connectHeaders);
        }
        this.changeState(Constants.ActivationState.ACTIVE);
    }

    /**
     * 启动断开，并设置客户端活动状态
     */
    private void deactivate() {
        if (isConnected()) {
            this.stompHandler.disconnect(this.disconnectHeaders);
        }
        this.changeState(Constants.ActivationState.INACTIVE);
    }

    /**
     * 改变客户端的活动状态
     * @param state {@link Constants.ActivationState} 状态
     */
    private void changeState(Constants.ActivationState state) {
        this.state = state;
    }
}
